package com.hsogoo.ultraman.monitor;

import java.rmi.Remote;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

import com.hsogoo.ultraman.listener.RemoteConfigChangeListener;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;


import com.hsogoo.ultraman.Constant;
import com.hsogoo.ultraman.event.ConfigChangeEvent;
import com.hsogoo.ultraman.listener.ConfigChangeListener;
import com.hsogoo.ultraman.listener.ConfigServerChangeListener;

import lombok.extern.slf4j.Slf4j;

/**
 * Created by za-huangsaigang on 2018/9/12.
 */
@Slf4j
public class ZkClient {

    private static class ZkClientHolder{
        private static final ZkClient instance = new ZkClient();
    }
    private static AtomicBoolean isRegistered = new AtomicBoolean(false);

    private static CuratorFramework client = null;

    private ZkClient(){}

    public synchronized static ZkClient getInstance(String url){
        if (client == null) {
            client = CuratorFrameworkFactory.builder()
                    .connectString(url)
                    .sessionTimeoutMs(1000)
                    .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                    .build();
            client.start();
        }
        return ZkClientHolder.instance;
    }

    /**
     * 创建根目录
     */
    public void createRootPath() {
        try {
            Stat stat = client.checkExists().forPath(Constant.CONFIG_ROOT_PATH);
            if (stat == null) {
                client.create().withMode(CreateMode.PERSISTENT).forPath(Constant.CONFIG_ROOT_PATH);
            }
        } catch (Exception e) {
            log.error("", e);
        }
    }

    /**
     * 创建节点
     * @param path
     * @param mode
     */
    public void createNode(String path, CreateMode mode) {
        try {
            if (mode == CreateMode.EPHEMERAL) {
                for (int i = 1; i <= 30; i++) {
                    Stat stat = client.checkExists().forPath(path);
                    if (stat == null) {
                        client.create().withMode(mode).forPath(path);
                        isRegistered.compareAndSet(false, true);
                        break;
                    }else if (!isRegistered.get()){
                        //第一次注册的时候需要处理临时节点消失延迟问题，检测30s
                        Thread.sleep(1000);
                        log.info("注册节点中..." + path);
                    }else{
                        return;
                    }
                }
            } else {
                Stat stat = client.checkExists().forPath(path);
                if (stat == null) {
                    client.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
                }
            }
        } catch (Exception e) {
            log.error("", e);
        }
    }

    /**
     * 写数据：/zookeeper/app
     * @param path
     * @param content
     *
     * @return 返回真正写到的路径
     * @throws Exception
     */
    public String write(String path,String content) throws Exception {
        StringBuilder sb = new StringBuilder(path);
        String writePath = client.create().creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                .forPath(sb.toString(), content.getBytes("utf-8"));
        return writePath;
    }

    public String read(String path) throws Exception {
        byte[] b = client.getData().forPath(path);
        String value = new String(b, "utf-8");
        return value;
    }


    /**
     * 监控配置服务器列表变化
     * @param path
     */
    public void watchServerList(final String path, final ConfigServerChangeListener configServerChangeListener) {
        try {
            PathChildrenCache childCache = new PathChildrenCache(client, path, false);
            //注册监听
            childCache.getListenable().addListener(new PathChildrenCacheListener(){
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    switch (event.getType()) {
                        case CHILD_ADDED: {
                            log.info("Child Node added: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                            break;
                        }
                        case CHILD_UPDATED: {
                            log.info("Child Node changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                            break;
                        }
                        case CHILD_REMOVED: {
                            log.info("Child Node removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                            break;
                        }
                    }
                    configServerChangeListener.onChange();
                }
            });
            childCache.start();
        } catch (Exception e) {
            log.error("", e);
        }
    }

    /**
     * 监控配置包括子节点被修改时刷新配置
     * @param path
     */
    public void watchWithChild(final String path, final RemoteConfigChangeListener remoteConfigChangeListener) {
        try {
            /** 只监听当前path的节点
            final NodeCache nodeChahe = new NodeCache(client, path);
            nodeChahe.getListenable().addListener(new NodeCacheListener() {

                public void nodeChanged() throws Exception {
                    log.info("nodeChanged:" + new String(nodeChahe.getCurrentData().getData(),"utf-8"));
                    ConfigRefresher.refresh(path);
                }
            });
            nodeChahe.start();**/

            /**只监听当前节点的第一层子节点
            PathChildrenCache childCache = new PathChildrenCache(client, path, false);
            //注册监听
            childCache.getListenable().addListener(new PathChildrenCacheListener(){
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    log.info("Child Node change...");
                    switch (event.getType()) {
                        case CHILD_ADDED: {
                            log.info("Child Node added: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                            break;
                        }
                        case CHILD_UPDATED: {
                            log.info("Child Node changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                            break;
                        }
                        case CHILD_REMOVED: {
                            log.info("Child Node removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                            break;
                        }
                    }
                }
            });
            childCache.start();**/


            //对path路径下所有孩子节点的监听
            TreeCache treeCache=new TreeCache(client, path);
            treeCache.getListenable().addListener(new TreeCacheListener(){

                @Override
                public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {
                    log.info("Tree Node change...");
                    String namespace = null;
                    switch (event.getType()) {
                        case INITIALIZED:
                            //doNothing
                            break;
                        case NODE_ADDED: {
                            log.info("Tree Node added: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                            log.info("Tree path : " + event.getData().getPath());
                            remoteConfigChangeListener.onChange(parseNamespaceFromNode());
                            break;
                        }
                        case NODE_UPDATED: {
                            log.info("Tree Node changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                            log.info("Tree path : " + event.getData().getPath());
                            remoteConfigChangeListener.onChange(parseNamespaceFromNode());
                            break;
                        }
                        case NODE_REMOVED: {
                            log.info("Tree Node removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
                            log.info("Tree path : " + event.getData().getPath());
                            remoteConfigChangeListener.onChange(parseNamespaceFromNode());
                            break;
                        }
                    }

                }

                private String parseNamespaceFromNode() {
                    //default
                    return "app";
                }

            });
            treeCache.start();
        } catch (Exception e) {
            log.error("", e);
        }
    }

    public List<String> getAllServer() {
        try {
            return client.getChildren().forPath(Constant.CONFIG_SERVER_LIST);
        } catch (Exception e) {
            log.error("", e);
        }
        return null;
    }

    public static void main(String[] args) {

    }
}
